package com.zidongxiangxi.rabbit.consumer.interceptor;

import com.zidongxiangxi.rabbit.consumer.manager.ConsumeRecordManager;
import com.zidongxiangxi.rabbit.core.util.RabbitUtils;
import org.aopalliance.intercept.MethodInvocation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.StringUtils;

import java.util.Objects;

/**
 * 幂等拦截器
 *
 * @author chenxudong
 * @date 2020/11/07
 */
public class IdempotentOperationsInterceptor extends AbstractConsumerOperationsInterceptor {
	private final Logger logger = LoggerFactory.getLogger(IdempotentOperationsInterceptor.class);

	private final String application;
	private final ConsumeRecordManager consumeRecordManager;
	private final TransactionTemplate transactionTemplate;

	public IdempotentOperationsInterceptor(
			String application,
			ConsumeRecordManager consumeRecordManager,
			TransactionTemplate transactionTemplate
	) {
		this.application = application;
		this.consumeRecordManager = consumeRecordManager;
		this.transactionTemplate = transactionTemplate;
	}

	@Override
	public Object invoke(MethodInvocation invocation) throws Throwable {
		Message message = getMessage(invocation);
		if (Objects.isNull(message)) {
			return invocation.proceed();
		}
		String messageId = RabbitUtils.getMessageId(message);
		if (StringUtils.isEmpty(messageId)) {
			return invocation.proceed();
		}
		String produceApplication = RabbitUtils.getProduceApplication(message);
		if (StringUtils.isEmpty(produceApplication)) {
			return invocation.proceed();
		}
		if (consumeRecordManager.isConsumed(produceApplication, application, messageId)) {
			logger.warn("消息[{}, {} ,{}]已经消费国, 忽略", produceApplication, application, message);
			return null;
		}
		if (Objects.nonNull(transactionTemplate)) {
			return transactionTemplate.execute((transactionStatus) -> {
				try {
					if (!consumeRecordManager.saveRecord(produceApplication, application, messageId)) {
						return null;
					}
					return invocation.proceed();
				} catch (Throwable throwable) {
					logger.error("消费失败", throwable);
					throw new RuntimeException(throwable);
				}
			});
		} else {
			try {
				if (!consumeRecordManager.saveRecord(produceApplication, application, messageId)) {
					return null;
				}
				return invocation.proceed();
			} catch (Throwable throwable) {
				logger.error("消费失败", throwable);
				consumeRecordManager.deleteRecord(produceApplication, application, messageId);
				throw new RuntimeException(throwable);
			}
		}
	}
}
